#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/function.hpp>
#include <boost/bind/placeholders.hpp>
#include <boost/system/system_error.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <boost/noncopyable.hpp>
#include <boost/array.hpp>
#include <boost/thread.hpp>
#include <iostream>
#include <vector>
#include <list>

using namespace boost::asio;

typedef boost::shared_ptr<ip::tcp::socket> socket_ptr;

class TestEchoConnection;

typedef boost::shared_ptr<TestEchoConnection> TestEchoConnection_ptr;

typedef std::list<TestEchoConnection_ptr> ConnectionManager;



template <typename T, typename T2>

class MutexList {
public:

    MutexList()
    {

    }

    ~MutexList()
    {

    }

    void remove(const T2& value)
    {
        m_mutex.lock();
        m_t2s.remove(value);
        m_mutex.unlock();
    }

    void push_back(const T2& value)
    {
        m_mutex.lock();
        m_t2s.push_back(value);
        m_mutex.unlock();
    }



private:
    std::list<T2> m_t2s;
    T m_mutex;

};

typedef MutexList<boost::mutex, TestEchoConnection_ptr> MutexConnectionManager;




#define CPUS sysconf(_SC_NPROCESSORS_ONLN)

class io_service_pool : private boost::noncopyable
{
public:

    typedef boost::shared_ptr<io_service> io_service_ptr;
    typedef boost::shared_ptr<io_service::work> work_ptr;

    io_service_pool()
        : m_index(0)
    {
        std::cout << "--init--" << std::endl;
        init();
        std::cout << "--init--" << std::endl;
    }

    ~io_service_pool()
    {

    }

    void start()
    {
        std::cout << "--start--" << std::endl;
        if (!m_threads.size()) {
            for (std::vector<io_service_ptr>::iterator ite = m_io_services.begin();
                 ite != m_io_services.end(); ++ite) {
                m_threads.create_thread(boost::bind(&io_service::run, (*ite).get()));
            }
            std::cout << "--join--" << std::endl;
            m_threads.join_all();
            std::cout << "--join--" << std::endl;
        }
        std::cout << "--start--" << std::endl;
    }

    void stop()
    {
        for (std::vector<io_service_ptr>::iterator ite = m_io_services.begin();
             ite != m_io_services.end(); ++ite) {
            (*ite)->post(boost::bind(&io_service::stop, (*ite).get()));
        }
        m_works.clear();
    }

    io_service_ptr get()
    {
        if (m_index >= m_io_services.size()) {
            m_index = 0;
        }

        return m_io_services[m_index++];
    }

private:
    void init(int cpu = CPUS)
    {
        std::cout << "--cpu--:" << cpu << std::endl;
        for (int i = 0; i < cpu; ++i) {
            m_io_services.push_back(io_service_ptr(new io_service));
            m_works.push_back(work_ptr(new io_service::work(*m_io_services.back())));
        }
    }

    int m_index;

    std::vector<io_service_ptr> m_io_services;
    std::vector<work_ptr> m_works;
    boost::thread_group m_threads;


};


class TestEchoConnection : public boost::enable_shared_from_this<TestEchoConnection>
        , private boost::noncopyable
{
public:
    TestEchoConnection(io_service &server, MutexConnectionManager &connectManager)
        : m_server(server)
        , m_connectManager(connectManager)
        , m_socket(new ip::tcp::socket(m_server))
    {

    }

    ~TestEchoConnection()
    {
        std::cout << "~TestEchoConnection" << std::endl;
    }

    ip::tcp::socket& socket()
    {
        return *m_socket;
    }

    void start()
    {
        m_socket->async_read_some(buffer(m_buffer), boost::bind(&TestEchoConnection::handle_read, shared_from_this(),
                                                                _1, _2));
    }


    void stop()
    {
        m_socket->close();
        m_connectManager.remove(TestEchoConnection_ptr(shared_from_this()));
    }


private:
    void handle_read(const boost::system::error_code &ec, size_t data_size)
    {
        if (!ec) {
            std::cout << "handle_read->data size:" << data_size << std::endl;
            m_socket->async_write_some(buffer(m_buffer), boost::bind(&TestEchoConnection::handler_write,
                                                                     shared_from_this(), _1));
//            m_socket->async_read_some(buffer(m_buffer), boost::bind(&TestEchoConnection::handle_read, shared_from_this(),
//                                                                    _1, _2));


        }
        else if (ec == error::operation_aborted || ec == error::eof) {
            std::cout << "handle_read" << "remove" << std::endl;
            stop();

        }
    }

    void handler_write(const boost::system::error_code &ec)
    {
        if (!ec) {
            memset(&m_buffer, 0, sizeof(m_buffer));
//            m_socket->async_read_some(buffer(m_buffer), boost::bind(&TestEchoConnection::handle_read, shared_from_this(),
//                                                                    _1, _2));
            if (ec == error::operation_aborted) {
                std::cout << "handler_write" << "remove" << std::endl;
                stop();

            }
        }
        stop();
    }




private:
    io_service& m_server;
    MutexConnectionManager& m_connectManager;
    socket_ptr m_socket;
    boost::array<char, 1029> m_buffer;


};


class TestEchoServer : public boost::enable_shared_from_this<TestEchoServer>
        , private boost::noncopyable
{
public:
    TestEchoServer(io_service &service, int port)
        : m_io_service(service)
        , m_endpoint(ip::tcp::v4(), port)
        , m_acceptor(m_io_service, m_endpoint)
    {

    }

    ~TestEchoServer()
    {

    }

    void start_accept()
    {
        m_connect.reset(new TestEchoConnection(m_io_service, m_connectionManager));
        m_acceptor.async_accept(m_connect->socket(), boost::bind(&TestEchoServer::handle_accept, shared_from_this(), _1));
    }

private:

    void handle_accept(const boost::system::error_code &err)
    {
        if (!err) {
            m_connectionManager.push_back(m_connect);
            m_connect->start();

        }
        start_accept();
    }


private:
    io_service& m_io_service;
    ip::tcp::endpoint m_endpoint;
    ip::tcp::acceptor m_acceptor;
    MutexConnectionManager m_connectionManager;
    TestEchoConnection_ptr m_connect;

};



class TestEchoServer2 : public boost::enable_shared_from_this<TestEchoServer2>
        , private boost::noncopyable
{
public:
    TestEchoServer2(io_service_pool &service_pool, int port)
        : m_io_service_pool(service_pool)
        , m_endpoint(ip::tcp::v4(), port)
        , m_acceptor(*(m_io_service_pool.get()), m_endpoint)
    {

    }

    ~TestEchoServer2()
    {

    }

    void start_accept()
    {
        m_connect.reset(new TestEchoConnection(*(m_io_service_pool.get()), m_connectionManager));
        m_acceptor.async_accept(m_connect->socket(), boost::bind(&TestEchoServer2::handle_accept, shared_from_this(), _1));
    }

private:

    void handle_accept(const boost::system::error_code &err)
    {
        if (!err) {
            m_connectionManager.push_back(m_connect);
            m_connect->start();

        }
        start_accept();
    }


private:
    io_service_pool& m_io_service_pool;
    ip::tcp::endpoint m_endpoint;
    ip::tcp::acceptor m_acceptor;
    MutexConnectionManager m_connectionManager;
    TestEchoConnection_ptr m_connect;

};









// echo server
int main()
{

    io_service_pool service_pool;
    boost::shared_ptr<TestEchoServer2> testserver2(new TestEchoServer2(service_pool, 8090));
//    TestEchoServer tesetserver(service, 8090);
//    tesetserver.start_accept();
    testserver2->start_accept();

    service_pool.start();
    return 0;

}



